package org.yuanzheng.tableAndSql

import org.apache.flink.api.common.typeinfo.TypeInformation
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.table.api.scala.StreamTableEnvironment
import org.apache.flink.table.api.{EnvironmentSettings, Table, Types}
import org.apache.flink.table.functions.TableFunction
import org.apache.flink.types.Row

object TestUDFByWordCount {
  def main(args: Array[String]): Unit = {
    //创建使用flink原生
    val streamEnv: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    val settings: EnvironmentSettings = EnvironmentSettings.newInstance().useOldPlanner().inStreamingMode().build()
    val tableEnv: StreamTableEnvironment = StreamTableEnvironment.create(streamEnv, settings)

    //隐式转换
    import org.apache.flink.streaming.api.scala._
    import org.apache.flink.table.api.scala._

    //读取数据
    val stream: DataStream[String] = streamEnv.socketTextStream("192.168.1.10", 8888)
    val table: Table = tableEnv.fromDataStream(stream, 'line)

    //使用tableApi切割单词，需要自定义一个切割单词的函数
    val flatMapFunction = new myFlatMapFunction
    val result: Table = table.flatMap(flatMapFunction('line)).as('word, 'wordCount).groupBy('word).select('word, 'wordCount.sum as 'count)

    tableEnv.toRetractStream[Row](result).filter(_._1 == true).print()
    tableEnv.execute("table_api")
    //streamEnv.execute()
  }
}

//自定义UDF
class myFlatMapFunction extends TableFunction[Row] {
  //输出单词和1
  override def getResultType: TypeInformation[Row] = Types.ROW(Types.STRING(), Types.INT())

  //函数主体
  def eval(str: String): Unit = {
    str.split(",").foreach(word => {
      val row = new Row(2)
      row.setField(0, word)
      row.setField(1, 1)
      collect(row)
    })
  }
}